package com.education.realtime.util;



import com.education.realtime.common.Constant;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;



public class FlinkSourceUtil {
    public static SourceFunction<String> getKafkaSource(String groupId, String topic) {

        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("group.id",groupId);
        props.put("auto.reset.offsets","latest");
        props.put("isolation.level","read_committed");//不读取未提交的数据
        //FlinkKafkaConsumer 已被弃用并将在 Flink 1.15 中移除，请改用 KafkaSource。
        return new FlinkKafkaConsumer<String>(
                topic,
                new SimpleStringSchema(),
                props
        );
    }


}
